package com.example

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}

case class Iris(id: Int,
                sepalLengthCm: Double,
                sepalWidthCm: Double,
                petalLengthCm: Double,
                petalWidthCm: Double,
                species: String)

object Test4 {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession
      .builder()
      .master("local[*]")
      .appName(this.getClass.getCanonicalName)
      .getOrCreate()
    val sc = spark.sparkContext
    sc.setLogLevel("warn")
    val K: Int = 5
    val irisRdd: RDD[Iris] = sc.textFile("data/Iris.csv")
      .map(line => {
        val strings: Array[String] = line.split(",")
        Iris(strings(0).toInt, strings(1).toDouble, strings(2).toDouble, strings(3).toDouble,
          strings(4).toDouble, strings(5))

      })

    val unknowIrisArray: Array[Iris] = sc.textFile("data/Iris-unknow.csv")
      .map(line => {
        val strings: Array[String] = line.split(",")
        Iris(strings(0).toInt, strings(1).toDouble, strings(2).toDouble, strings(3).toDouble,
          strings(4).toDouble, null)
      }).collect()

    unknowIrisArray.foreach(println(_))

    println("==========================")

    unknowIrisArray.foreach(unknowIris => {

      val speciesMap: Map[String, Int] = irisRdd.map(iris => {
        val distance = getDistance(unknowIris, iris)
        (distance, iris)
      })
        // 通过距离进行排序
        .sortByKey()
        // 取前K个
        .take(K)
        .map(item => {
          item._2.species
        }).groupBy(x => x)
        .mapValues(_.length)

      import scala.collection.immutable.ListMap
      val resultMap: ListMap[String, Int] = ListMap(speciesMap.toSeq.sortBy(_._2): _*)

      println(Iris(unknowIris.id,
              unknowIris.sepalLengthCm, unknowIris.sepalWidthCm,
              unknowIris.petalLengthCm, unknowIris.petalWidthCm,
        resultMap.keys.toList.head))
    })
    sc.stop()

  }




  def getDistance(x: Iris, y: Iris): Double = {
    import scala.math._

    sqrt(
      square(x.sepalWidthCm - y.sepalWidthCm)
        + square(x.petalLengthCm - y.sepalLengthCm)
        + square(x.petalWidthCm - y.petalWidthCm)
        + square(x.petalLengthCm - y.petalLengthCm))
  }

  def square(x: Double) = x * x


}


